package com.queue.cli.client;

import com.queue.cli.handler.ConsumerHandler;
import com.queue.cli.initializer.ConsumerInitializer;
import com.queue.cli.listener.MessageData;
import com.queue.cli.listener.MessageListener;
import com.queue.cli.util.ResourceUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Consumer {
    //服务器地址
    private static String host;

    //服务器端口
    private static int port;

    //消息处理器
    private ConsumerHandler consumerHandler;

    //当前消费者的通道
    private Channel channel;

    //日志信息
    private Logger logger = LoggerFactory.getLogger(Consumer.class);

    static {
        host = ResourceUtils.getKey("server.host");
        port = Integer.parseInt(ResourceUtils.getKey("server.port"));
    }

    /**
     * @param destination 队列/主题名
     */
    public Consumer(String destination) {
        consumerHandler = new ConsumerHandler(destination);
    }

    private void run() {
        //检测消息监听器是否启动
        if (this.consumerHandler.getMessageListener() == null) {
            throw new NullPointerException("未设置消息监听器...");
        }

        NioEventLoopGroup clientGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(clientGroup).channel(NioSocketChannel.class).handler(new ConsumerInitializer(consumerHandler));
        try {
            //连接服务器
            ChannelFuture sync = bootstrap.connect(host, port).sync();
            this.channel = sync.channel();
            //监听关闭连接
            sync.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            //关闭工作组
            clientGroup.shutdownGracefully();
        }

    }

    /**
     * 设置自动签名
     * @param autoAcknowledge
     */
    public void setAutoAcknowledge(boolean autoAcknowledge){
        this.consumerHandler.setAutoAcknowledge(autoAcknowledge);
    }

    /**
     * 创建消息监听器
     * @param listener
     * @return
     */
    public Consumer createListener(MessageListener listener){
        this.consumerHandler.createListener(listener);
        return this;
    }

    /**
     * 消费者示例
     * @param args
     */
    public static void main(String[] args) {
        //指定queue队列
        Consumer consumer = new Consumer("queue");
        //构建监听器
        consumer.createListener(new MessageListener() {
            @Override
            public void getMessage(MessageData data) {
                //监听到消息就会进入MessageListener监听器中
                System.out.println(data.getMessage());
            }
        }).run();

    }

}
